package spring.cloud.tasks.executor.task;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.executor.executor.ExecutorIdToTaskIdToTaskExecutorMapMap;
import spring.cloud.tasks.executor.executor.TaskInitService;
import spring.cloud.tasks.executor.listener.ListenerHolderManager;
import spring.cloud.tasks.executor.listener.analyse.AnalyseService;
import spring.cloud.tasks.executor.listener.config.TaskConfiguration;
import spring.cloud.tasks.executor.listener.config.TaskConfigurationService;
import spring.cloud.tasks.executor.listener.election.LeaderElectionService;
import spring.cloud.tasks.executor.listener.failover.FailoverService;
import spring.cloud.tasks.executor.listener.report.ReportService;
import spring.cloud.tasks.executor.listener.sharding.ShardingService;
import spring.cloud.tasks.executor.listener.task_execute.TaskExecutorInfoService;
import spring.cloud.tasks.executor.node.TaskNodeStorage;
import spring.cloud.tasks.executor.task.statistics.ProcessCountTaskService;
import spring.cloud.tasks.executor.task_type.TaskTypeManager;
import spring.cloud.tasks.executor.trigger.Scheduler;
import spring.cloud.tasks.executor.utils.ThreadFactory;
import spring.cloud.tasks.executor.utils.log.LogUtils;
import spring.cloud.tasks.executor.utils.zookeeper.NodeCacheOrTreeCacheManager;
import spring.cloud.tasks.executor.utils.zookeeper.Zookeeper;

import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
@Setter
@Getter
public class TaskExecutor {
    private final TaskConfiguration taskConfiguration;
    private final Zookeeper zookeeper;
    private final ListenerHolderManager listenerManager;
    private final TaskConfigurationService taskConfigurationService;
    private final LeaderElectionService leaderElectionService;
    private final TaskExecutorInfoService taskExecutorInfoService;
    private final ReportService reportService;
    private final ShardingService shardingService;
    private final TaskExecuteContextService executeContextService;
    private final ExecuteService executeService;
    private final FailoverService failoverService;
    private final ProcessCountTaskService processCountTaskService;
    private final AnalyseService analyseService;

    private final TaskNodeStorage taskNodeStorage;
    private final NodeCacheOrTreeCacheManager nodeCacheOrTreeCacheManager;
    ExecutorService executorService;
    private String taskId;
    private String executorId;
    /**
     * since all the conf-node values will be gotten from zk-cache. use this to compare with the new values.
     */

    private TaskConfiguration previousJobConfiguration = new TaskConfiguration(null, null);
    private TaskExecuteEngine taskExecuteEngine;

    private TaskInitService taskInitService;

    private AtomicBoolean isShutdownFlag = new AtomicBoolean(false);

    public TaskExecutor(final Zookeeper zookeeper, final TaskConfiguration taskConfiguration) {
        this.taskId = taskConfiguration.getTaskId();
        this.executorId = zookeeper.getExecutorId();
        this.taskConfiguration = taskConfiguration;
        this.zookeeper = zookeeper;
        this.taskNodeStorage = new TaskNodeStorage(zookeeper, taskConfiguration);
        initExecutorService();
        ExecutorIdToTaskIdToTaskExecutorMapMap.addTaskExecutor(executorId, taskId, this);

        nodeCacheOrTreeCacheManager = new NodeCacheOrTreeCacheManager(zookeeper.getCuratorFramework(), executorId, taskId);
        taskConfigurationService = new TaskConfigurationService(this);
        leaderElectionService = new LeaderElectionService(this);
        taskExecutorInfoService = new TaskExecutorInfoService(this);
        shardingService = new ShardingService(this);
        executeContextService = new TaskExecuteContextService(this);
        executeService = new ExecuteService(this);
        failoverService = new FailoverService(this);
        processCountTaskService = new ProcessCountTaskService(this);
        analyseService = new AnalyseService(this);

        listenerManager = new ListenerHolderManager(this);
        reportService = new ReportService(this);

        // see EnabledPathListener and CronPathListener, only these values are supposed to be watched.

        previousJobConfiguration.setCronExpression(taskConfiguration.getCronExpression());
        previousJobConfiguration.setPauseDatePeriod(taskConfiguration.getPauseDatePeriod());
        previousJobConfiguration.setPauseTimePeriod(taskConfiguration.getPauseTimePeriod());
        previousJobConfiguration.setProcessCountIntervalSeconds(taskConfiguration.getProcessCountIntervalSeconds());
    }

    /**
     * 初始化作业.
     */
    public void init() {
        try {
            startAll();
            createTaskExecuteEngine();
            taskExecutorInfoService.persistExecutorOnline();
            // Notify job enabled or disabled after that all are ready, include job was initialized.
            taskConfigurationService.notifyJobEnabledOrNot();
        } catch (Throwable t) {
            shutdown(false);
            throw t;
        }
    }

    private void startAll() {
        taskConfigurationService.start();
        leaderElectionService.start();
        taskExecutorInfoService.start();
        shardingService.start();
        executeContextService.start();
        executeService.start();
        failoverService.start();
        processCountTaskService.start();

        analyseService.start();


        listenerManager.start();
        leaderElectionService.leaderElection();

        taskExecutorInfoService.clearRunOneTimePath();
        taskExecutorInfoService.clearStopOneTimePath();
        taskExecutorInfoService.resetCount();
        processCountTaskService.startProcessCountTask();
    }

    private void createTaskExecuteEngine() {
        taskExecuteEngine = new TaskExecuteEngine();
        taskExecuteEngine.setTaskExecutor(this);
        taskExecuteEngine.setTaskConfigurationService(taskConfigurationService);
        taskExecuteEngine.setShardingService(shardingService);
        taskExecuteEngine.setExecuteContextService(executeContextService);
        taskExecuteEngine.setExecuteService(executeService);
        taskExecuteEngine.setFailoverService(failoverService);
        taskExecuteEngine.setTaskExecutorInfoService(taskExecutorInfoService);
        taskExecuteEngine.setExecutorId(executorId);
        taskExecuteEngine.setReportService(reportService);
        taskExecuteEngine.setTaskId(taskId);
        taskExecuteEngine.setNameSpace(zookeeper.getNameSpace());
        taskExecuteEngine.setTaskInitService(taskInitService);
        taskExecuteEngine.init();
    }

    private void initExecutorService() {
        java.util.concurrent.ThreadFactory threadFactory = new ThreadFactory(taskId);
        executorService = Executors.newFixedThreadPool(Integer.MAX_VALUE, threadFactory);
    }

    public void reCreateExecutorService() {
        synchronized (isShutdownFlag) {
            if (isShutdownFlag.get()) {
                LogUtils.warn(log, taskId, "the jobScheduler was shutdown, cannot re-create business thread pool");
                return;
            }
            executeService.stop();
            initExecutorService();
        }
    }

    public void shutdownExecutorService() {
        if (executorService != null && !executorService.isShutdown()) {
            executorService.shutdown();
        }
    }

    /**
     * 获取下次作业触发时间.可能被暂停时间段所影响。
     *
     * @return 下次作业触发时间
     */
    public Date getNextFireTimePausePeriodEffected() {
        try {
            Scheduler scheduler = taskExecuteEngine.getScheduler();
            return scheduler == null ? null : scheduler.getNextFireTimePausePeriodEffected();
        } catch (Throwable t) {
            LogUtils.error(log, taskId, "fail to get next fire time", t);
            return null;
        }
    }

    /**
     * 立刻启动作业.
     */
    public void triggerTask() {
        if (taskExecuteEngine.getScheduler().isShutdown()) {
            return;
        }
        taskExecuteEngine.getScheduler().trigger();
    }

    /**
     * 关闭process count thread
     */
    public void shutdownCountThread() {
        processCountTaskService.stop();
    }

    public void shutdown(boolean removeJob) {
        synchronized (isShutdownFlag) {
            isShutdownFlag.set(true);

            // 关闭Listener
            listenerManager.stop();

            // 关闭作业：关闭调度器，强杀业务
            if (taskExecuteEngine != null) {
                taskExecuteEngine.shutdown();
            }

            // 关闭服务
            shardingService.stop();
            taskConfigurationService.stop();
            leaderElectionService.stop();
            taskExecutorInfoService.stop();
            executeContextService.stop();
            executeService.stop();
            failoverService.stop();
            processCountTaskService.stop();
            analyseService.stop();


            // 关闭TreeCache
            nodeCacheOrTreeCacheManager.shutdown();

            // 删除作业
            if (removeJob) {
                taskNodeStorage.deleteJobNode();
                taskInitService.removeTaskId(taskId);
            }

            // 移除作业注册表
            ExecutorIdToTaskIdToTaskExecutorMapMap.removeTaskExecutor(executorId, taskId);
        }
    }

    /**
     * 重新初始化Trigger
     */
    public void reInitializeTrigger() {
        taskExecuteEngine.getScheduler().reInitializeTrigger();
    }

    /**
     * 重启统计处理数据数量的任务
     */
    public void rescheduleProcessCountJob() {
        processCountTaskService.startProcessCountTask();
    }

    public boolean isAllowedShutdownGracefully() {
        return TaskTypeManager.get(taskConfiguration.getTaskType()).isAllowedShutdownGracefully();
    }


}
